package com.yangji.rxjavatest.rx.observable;

import com.yangji.rxjavatest.Log;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/*
 * http://www.jianshu.com/p/128e662906af
 * flatMap 无序
 * concatMap 有序
 */
public class RxJavaFlatMap {

    private static final String TAG = "MainActivity";

    public RxJavaFlatMap() {
        //testFlatMap();


        //下面模拟用户注册后，继续用户登录
        register()
                .doOnNext(new Consumer<RegisterInfo>() {
                    @Override
                    public void accept(RegisterInfo registerInfo) throws Exception {
                        //先根据注册的响应结果去做一些操作
                        //比如这里就是成功 给UI提示注册成功
                        Log.d(TAG, "注册成功！" + Thread.currentThread().getName());
                    }
                })
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.d(TAG, "注册失败！" + Thread.currentThread().getName());
                    }
                })
                .flatMap(new Function<RegisterInfo, ObservableSource<LoginInfo>>() {
                    @Override
                    public ObservableSource<LoginInfo> apply(RegisterInfo registerInfo) throws Exception {
                        Log.d(TAG, "注册成功！前往登录" + Thread.currentThread().getName());
                        //用户注册成功 马上调用用户登录去登录

                        if(registerInfo == null){
                            return null;
                        }

                        return login(registerInfo.userName, registerInfo.userPwd);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求登录的结果
                .doOnError(new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        Log.d(TAG, "登录失败！" + Thread.currentThread().getName());
                    }
                })
                .subscribe(new Consumer<LoginInfo>() {
                    @Override
                    public void accept(LoginInfo o) throws Exception {
                        //这里处理用户登录成功
                        Log.d(TAG, "登录成功:" + o.getClass().getName() + " " + Thread.currentThread().getName());
                    }
                });
    }

    //比如这里是用户注册
    private Observable<RegisterInfo> register() {
        return Observable
                .create(new ObservableOnSubscribe<RegisterInfo>() {
                    @Override
                    public void subscribe(ObservableEmitter<RegisterInfo> e) throws Exception {
                        //假如这里用户注册成功
                        Log.d(TAG, "来注册:" + Thread.currentThread().getName());
//                        e.onNext(new RegisterInfo("admin", "123456"));
//                        e.onComplete();
                        e.onError(new NullPointerException("注册失败"));
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    //比如这里是用户登录
    private Observable<LoginInfo> login(String userName, String pwd) {
        return Observable
                .create(new ObservableOnSubscribe<LoginInfo>() {
                    @Override
                    public void subscribe(ObservableEmitter<LoginInfo> e) throws Exception {
                        Log.d(TAG, "来登录:" + Thread.currentThread().getName());
                        e.onNext(new LoginInfo());
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    class RegisterInfo {
        public String userName;
        public String userPwd;

        public RegisterInfo(String userName, String userPwd) {
            this.userName = userName;
            this.userPwd = userPwd;
        }
    }

    class LoginInfo {
    }

    public void testFlatMap() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            }
        }).doOnNext(new Consumer<Integer>() {
            @Override
            public void accept(@NonNull Integer integer) throws Exception {
                //这里可以处理数据源比如注册完成后 下面flatmap 可以处理注册完成后自动登录


            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    list.add("I am value " + integer);
                }
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, s);
            }
        });
    }

}
